package myapps5;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * consumer 异步手动提交位移 ，减少重复消费的问题出现，但仍然会出现重复消费
 */
public class SimpleConsumer {
	private static final String GROUPID = "groupA";
	private static final String topic = "KAFKA_TEST";
	private static AtomicBoolean running =new AtomicBoolean(true);
	public static void main(String[] args) {
		Properties props = new Properties();
		props.put("bootstrap.servers", "192.168.0.150:9092");
		props.put("group.id", GROUPID);
//		props.put("enable.auto.commit", "true");
//		props.put("auto.commit.interval.ms", "1000");
		//设置手动提交
		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
		props.put("session.timeout.ms", "30000");
		props.put("auto.offset.reset", "earliest");
		props.put("key.deserializer", StringDeserializer.class.getName());
		props.put("value.deserializer", StringDeserializer.class.getName());
		KafkaConsumer consumer = new KafkaConsumer<String, String>(props);
//		consumer.subscribe(Arrays.asList(topic));
		//指定分区（可选）
		TopicPartition tp = new TopicPartition(topic, 0);
		consumer.assign(Arrays.asList(tp));
		long offset = -1L;

		try{
			while (running.get()){
				ConsumerRecords<String,String> records = consumer.poll(1000);
				for (ConsumerRecord<String, String> record : records) {

				}
				//异步回调
				consumer.commitAsync(new OffsetCommitCallback() {
					@Override
					public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
						if(exception == null){
							System.out.println(offsets);
						}else {
							System.out.println("fail to commit offsets :"+ offsets+"--"+ exception.getMessage());
						}
					}
				});

			}
		}finally {
			consumer.close();
		}
	}
}
